package com.flinkjar.flinkjardemo.common;


import com.alibaba.fastjson.JSONObject;
import com.flinkjar.flinkjardemo.bean.UserProduct;
import com.flinkjar.flinkjardemo.config.DBConfig;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import io.debezium.data.Envelope;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Objects;

public class CustomSinkFunction extends RichSinkFunction<UserProduct> {

    /**
     * 日志
     */
    private static final Logger logger = LoggerFactory.getLogger(CustomSinkFunction.class);

    /**
     * 数据库连接池
     */
    private HikariDataSource dataSource;

    /**
     * 分表数量
     */
    private Integer tableCount = DBConfig.tableCount;


    /**
     * @param tableCount 分表数量
     */
    public CustomSinkFunction(Integer tableCount) {
        if (Objects.isNull(tableCount) || tableCount <= 0) {
            tableCount = DBConfig.tableCount;
        }
        this.tableCount = tableCount;
    }


    /**
     * open() 方法会在每个并行的 Sink 算子实例初始化时被调用一次。具体来说，当 Flink 作业启动时，每个 Sink 算子的实例都会被创建并初始化，这时 open() 方法会被调用。
     * 如果 Sink 算子的并行度为 n，则会创建 n 个并行的 Sink 算子实例，
     * 每个实例都会调用一次 open() 方法。
     * <p>
     * open() 方法用于执行一些初始化操作，如建立到外部系统的连接、创建资源等。
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        logger.info("进入open方法.........");

        // 懒加载初始化连接池
        if (dataSource == null) {
            synchronized (this) {
                if (dataSource == null) {
                    // eg: jdbc:mysql://1.92.97.72:3306/flink_test_db?useAffectedRows=true&useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai&allowMultiQueries=true
                    StringBuilder url = new StringBuilder();
                    url.append("jdbc:mysql://").append(DBConfig.hostname_sink).append(":").append(DBConfig.port_sink).append("/").append(DBConfig.database_sink).append("?useAffectedRows=true&useUnicode=true&characterEncoding=utf-8&useSSL=false&connectionTimeZone=Asia/Shanghai&allowMultiQueries=true");
                    HikariConfig config = new HikariConfig();
                    // 指定驱动
                    config.setDriverClassName("com.mysql.cj.jdbc.Driver");
                    config.setJdbcUrl(url.toString());
                    config.setUsername(DBConfig.username_sink);
                    config.setPassword(DBConfig.password_sink);
                    // 当池达到这个大小，并且没有空闲连接可用时，对getConnection()的调用将在超时之前阻塞长达connectionTimeout毫秒
                    config.setMaximumPoolSize(5);
                    config.setConnectionTimeout(2000L);
                    // 此属性控制池中连接的最大生存期。当连接达到此超时时，即使最近使用过，它也将从池中退役。正在使用的连接永远不会退役，只有当它空闲时才会被删除。
                    config.setMaxLifetime(1000L);
                    // 该属性控制HikariCP试图在池中维护的空闲连接的最小数量，包括空闲连接和正在使用的连接。如果空闲连接低于此值，HikariCP将尽最大努力快速有效地恢复它们。
                    config.setMinimumIdle(1);
                    dataSource = new HikariDataSource(config);

                    logger.info("数据库连接池创建成功.........");
                }
            }
        }

    }

    /**
     * 写入数据库的逻辑
     * 每来一数据就执行一次
     * TODO 这里也可以优化一下,使用状态将数据攒批后输出,这也可以调高吞吐量
     *
     * @param value
     * @param context
     */
    @Override
    public void invoke(UserProduct value, Context context) {
        // 获取表下标
        int tableIndex = Math.abs(value.getUserId().hashCode()) % tableCount;
        logger.info("接收到数据[index={}],[value={}]", tableIndex, JSONObject.toJSONString(value));

        // 获取sql模版
        String sqlTemplate = SqlTemplateService.getInsertSqlTemplate(tableIndex);
        // 获取操作类型
        Envelope.Operation operation = value.getOperation();
        if (Envelope.Operation.UPDATE.equals(operation)) {
            sqlTemplate = SqlTemplateService.getUpdateSqlTemplate(tableIndex);
        } else if (Envelope.Operation.DELETE.equals(operation)) {
            sqlTemplate = SqlTemplateService.getDeleteSqlTemplate(tableIndex);
        }

        try (// 获取数据库连接
             Connection connection = dataSource.getConnection();
             // 获取预编译语句
             PreparedStatement preparedStatement = connection.prepareStatement(sqlTemplate)) {

            // 禁用事务
            connection.setAutoCommit(true);

            // 写入mysql
            toMysql(value, preparedStatement);

        } catch (Exception e) {
            logger.error("数据写入异常, value={}, 异常信息:{}", value, e.getMessage(), e);
            e.printStackTrace();
        }

    }

    /**
     * 写入mysql
     *
     * @param value             不可为空
     * @param preparedStatement 不可为空
     * @throws SQLException
     */
    private void toMysql(UserProduct value, PreparedStatement preparedStatement) throws SQLException {
        // 获取操作类型
        Envelope.Operation operation = value.getOperation();

        if (Envelope.Operation.CREATE.equals(operation) || Envelope.Operation.READ.equals(operation)) {// 新增数据
            logger.info("新增数据, id={}", value.getId());
            if (Objects.isNull(preparedStatement)) {
                logger.error("新增数据失败, id={}", value.getId());
                return;
            }
            // insert into user_product_1 (id, user_id, product_id)  values  (?, ?, ?)
            preparedStatement.setLong(1, value.getId());
            preparedStatement.setString(2, value.getUserId());
            preparedStatement.setString(3, value.getProductId());
            // 执行sql
            preparedStatement.execute();
            logger.info("新增数据, sql={}", preparedStatement);

        } else if (Envelope.Operation.UPDATE.equals(operation)) {// 更新数据
            logger.info("更新数据, id={}", value.getId());
            if (Objects.isNull(preparedStatement)) {
                logger.error("更新数据失败, id={}", value.getId());
                return;
            }
            //update user_product_1 set  user_id = ?, product_id = ? where id = ? and userId = ?
            preparedStatement.setString(1, value.getUserId());
            preparedStatement.setString(2, value.getProductId());
            preparedStatement.setLong(3, value.getId());
            preparedStatement.setString(4, value.getUserId());
            // 执行sql
            preparedStatement.execute();
            logger.info("更新数据, sql={}", value);

        } else if (Envelope.Operation.DELETE.equals(operation)) {// 删除数据
            logger.info("删除数据, id={}", value.getId());
            if (Objects.isNull(preparedStatement)) {
                logger.error("删除数据失败, id={}", value.getId());
                return;
            }
            //delete from user_product_1 where id = ? and userId = ?
            preparedStatement.setLong(1, value.getId());
            preparedStatement.setString(2, value.getUserId());
            // 执行sql
            preparedStatement.execute();
            logger.info("删除数据, sql={}", value);

        }

    }


    /**
     * 任务结束时关闭资源
     *
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        super.close();

        // 关闭连接池
        if (dataSource != null) {
            dataSource.close();
        }

        logger.info("资源关闭成功.........");
    }


}
